package org.nbict.iot.protocol.trident;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.builtin.Count;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.trident.testing.MemoryMapState;
import org.apache.storm.trident.testing.Split;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

/**
 * Created by songseven on 18/6/29.
 */
public class BasicTridentTopology {
    public static void main(String[] args) {
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 3,
                new Values("this is simple example of trident topology"), new Values(
                "this example count same words"));
        //spout.setCycle(true);

        TridentTopology topology = new TridentTopology();
        MemoryMapState.Factory stateFactory = new MemoryMapState.Factory();
        Stream stream = topology
                .newStream("spout1", spout)
                .each(new Fields("sentence"), new Split(), new Fields("word"))
                .groupBy(new Fields("word")).persistentAggregate
                        (stateFactory, new Count(),
                        new Fields("count")).parallelismHint(6)
                .newValuesStream()
                .filter(new DisplayOutputFilter());

        Config config = new Config();
        config.setNumWorkers(1);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("storm-trident-example", config, topology.build());
    }
}
